#!/bin/bash
# ODS拉链表Relationship
# 因为需求需要将customer_relationship更新数据涉及到的维度重新统计；同时要有历史快照。
# 推荐采用拉链表(SCD2)的形式来做。需要在start_time字段的基础上，增加新的end_time字段，以标识封链时间。

HIVE_HOME=/usr/bin/hive
SQOOP_HOME=/usr/bin/sqoop

# 需求：能够手动传入参数指定采集的日期。如果不传入参数要求能够自动的识别前一天的日期
if [ $# -eq 0 ]
then
    DATESTR=`date -d "-1 day" +%Y-%m-%d`
else
    DATESTR=$1
fi

# 一、重建customer_relationship_update增量表，每次使用都要重建，避免数据重复问题
${HIVE_HOME} -e "
DROP TABLE IF EXISTS itcast_ods.customer_relationship_update;
CREATE TABLE IF NOT EXISTS itcast_ods.customer_relationship_update (
    id                          INT COMMENT '客户关系id',
    create_date_time            STRING COMMENT '创建时间',
    update_date_time            STRING COMMENT '最后更新时间',
    deleted                     INT COMMENT '是否被删除（禁用）',
    customer_id                 INT COMMENT '所属客户id',
    first_id                    INT COMMENT '第一条客户关系id',
    belonger                    INT COMMENT '归属人',
    belonger_name               STRING COMMENT '归属人姓名',
    initial_belonger            INT COMMENT '初始归属人',
    distribution_handler        INT COMMENT '分配处理人',
    business_scrm_department_id INT COMMENT '归属部门',
    last_visit_time             STRING COMMENT '最后回访时间',
    next_visit_time             STRING COMMENT '下次回访时间',
    origin_type                 STRING COMMENT '数据来源',
    itcast_school_id            INT COMMENT '校区Id',
    itcast_subject_id           INT COMMENT '学科Id',
    intention_study_type        STRING COMMENT '意向学习方式',
    anticipat_signup_date       STRING COMMENT '预计报名时间',
    level                       STRING COMMENT '客户级别',
    creator                     INT COMMENT '创建人',
    current_creator             INT COMMENT '当前创建人：初始==创建人，当在公海拉回时为 拉回人',
    creator_name                STRING COMMENT '创建者姓名',
    origin_channel              STRING COMMENT '来源渠道',
    comment                     STRING COMMENT '备注',
    first_customer_clue_id      INT COMMENT '第一条线索id',
    last_customer_clue_id       INT COMMENT '最后一条线索id',
    process_state               STRING COMMENT '处理状态',
    process_time                STRING COMMENT '处理状态变动时间',
    payment_state               STRING COMMENT '支付状态',
    payment_time                STRING COMMENT '支付状态变动时间',
    signup_state                STRING COMMENT '报名状态',
    signup_time                 STRING COMMENT '报名时间',
    notice_state                STRING COMMENT '通知状态',
    notice_time                 STRING COMMENT '通知状态变动时间',
    lock_state                  STRING COMMENT '锁定状态',
    lock_time                   STRING COMMENT '锁定状态修改时间',
    itcast_clazz_id             INT COMMENT '所属ems班级id',
    itcast_clazz_time           STRING COMMENT '报班时间',
    payment_url                 STRING COMMENT '付款链接',
    payment_url_time            STRING COMMENT '支付链接生成时间',
    ems_student_id              INT COMMENT 'ems的学生id',
    delete_reason               STRING COMMENT '删除原因',
    deleter                     INT COMMENT '删除人',
    deleter_name                STRING COMMENT '删除人姓名',
    delete_time                 STRING COMMENT '删除时间',
    course_id                   INT COMMENT '课程ID',
    course_name                 STRING COMMENT '课程名称',
    delete_comment              STRING COMMENT '删除原因说明',
    close_state                 STRING COMMENT '关闭装填',
    close_time                  STRING COMMENT '关闭状态变动时间',
    appeal_id                   INT COMMENT '申诉id',
    tenant                      INT COMMENT '租户',
    total_fee                   DECIMAL COMMENT '报名费总金额',
    belonged                    INT COMMENT '小周期归属人',
    belonged_time               STRING COMMENT '归属时间',
    belonger_time               STRING COMMENT '归属时间',
    transfer                    INT COMMENT '转移人',
    transfer_time               STRING COMMENT '转移时间',
    follow_type                 INT COMMENT '分配类型，0-自动分配，1-手动分配，2-自动转移，3-手动单个转移，4-手动批量转移，5-公海领取',
    transfer_bxg_oa_account     STRING COMMENT '转移到博学谷归属人OA账号',
    transfer_bxg_belonger_name  STRING COMMENT '转移到博学谷归属人OA姓名',
    end_time                    STRING COMMENT '有效时间'
) COMMENT '客户关系表'
PARTITIONED BY (start_time STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS ORC
TBLPROPERTIES ('orc.compress' = 'ZLIB');
"

# 模拟增加2条，修改1条
mysql -uroot -p123456 scrm -e "
INSERT INTO scrm.customer_relationship VALUES ('1', '2021-01-18 01:01:01', '2021-01-18 01:01:01', '', '2', '2', null, null, null, null, null, '2021-01-18 09:24:57', '2021-01-18 09:24:57', 'NETSERVICE', '1', null, null, '2021-01-18', 'A', '451', '451', '高因', '', '在职，只学习冲刺班框架第一个月的课程；问具体的开课时间；', '2', '2', null, null, null, null, null, null, null, null, '', null, null, null, null, null, null, null, null, null, null, null, null, null, 'RELEASE', null, null, '1', null, null, null, null, null, null, '1', null, null);
INSERT INTO scrm.customer_relationship VALUES ('3', '2021-01-18 03:03:03', '2021-01-18 03:03:03', '', '2', '2', null, null, null, null, null, '2021-01-18 09:24:57', '2021-01-18 09:24:57', 'NETSERVICE', '1', null, null, '2021-01-18', 'A', '451', '451', '高因', '', '在职，只学习冲刺班框架第一个月的课程；问具体的开课时间；', '2', '2', null, null, null, null, null, null, null, null, '', null, null, null, null, null, null, null, null, null, null, null, null, null, 'RELEASE', null, null, '1', null, null, null, null, null, null, '1', null, null);
UPDATE scrm.customer_relationship SET update_date_time = '2021-01-18 10:00:00' WHERE id = 2;
UPDATE scrm.customer_relationship SET belonger_name = '浪费时间啊'  WHERE id =2;
"

# 二、抽取昨天新增和更新的数据
# -m的个数为1：https://blog.csdn.net/iknownothinglee/article/details/102937775
${SQOOP_HOME} import \
--connect jdbc:mysql://192.168.10.150:3306/scrm \
--username root \
--password 123456 \
--query "
SELECT *,
       '9999-12-31' AS end_time,
       '${DATESTR}' AS start_time
FROM customer_relationship
WHERE (
        create_date_time >= '${DATESTR} 00:00:00'
        AND
        create_date_time <= '${DATESTR} 23:59:59')
   OR (
        update_date_time >= '${DATESTR} 00:00:00'
        AND
        update_date_time <= '${DATESTR} 23:59:59') AND \$CONDITIONS" \
--hcatalog-database itcast_ods \
--hcatalog-table customer_relationship_update \
--hive-partition-key start_time \
--hive-partition-value ${DATESTR} \
-m 1 \
--split-by id

# 三、重建customer_relationship_tmp临时表
${HIVE_HOME} -e "
DROP TABLE IF EXISTS itcast_ods.customer_relationship_tmp;
CREATE TABLE IF NOT EXISTS itcast_ods.customer_relationship_tmp (
    id                          INT COMMENT '客户关系id',
    create_date_time            STRING COMMENT '创建时间',
    update_date_time            STRING COMMENT '最后更新时间',
    deleted                     INT COMMENT '是否被删除（禁用）',
    customer_id                 INT COMMENT '所属客户id',
    first_id                    INT COMMENT '第一条客户关系id',
    belonger                    INT COMMENT '归属人',
    belonger_name               STRING COMMENT '归属人姓名',
    initial_belonger            INT COMMENT '初始归属人',
    distribution_handler        INT COMMENT '分配处理人',
    business_scrm_department_id INT COMMENT '归属部门',
    last_visit_time             STRING COMMENT '最后回访时间',
    next_visit_time             STRING COMMENT '下次回访时间',
    origin_type                 STRING COMMENT '数据来源',
    itcast_school_id            INT COMMENT '校区Id',
    itcast_subject_id           INT COMMENT '学科Id',
    intention_study_type        STRING COMMENT '意向学习方式',
    anticipat_signup_date       STRING COMMENT '预计报名时间',
    level                       STRING COMMENT '客户级别',
    creator                     INT COMMENT '创建人',
    current_creator             INT COMMENT '当前创建人：初始==创建人，当在公海拉回时为 拉回人',
    creator_name                STRING COMMENT '创建者姓名',
    origin_channel              STRING COMMENT '来源渠道',
    comment                     STRING COMMENT '备注',
    first_customer_clue_id      INT COMMENT '第一条线索id',
    last_customer_clue_id       INT COMMENT '最后一条线索id',
    process_state               STRING COMMENT '处理状态',
    process_time                STRING COMMENT '处理状态变动时间',
    payment_state               STRING COMMENT '支付状态',
    payment_time                STRING COMMENT '支付状态变动时间',
    signup_state                STRING COMMENT '报名状态',
    signup_time                 STRING COMMENT '报名时间',
    notice_state                STRING COMMENT '通知状态',
    notice_time                 STRING COMMENT '通知状态变动时间',
    lock_state                  STRING COMMENT '锁定状态',
    lock_time                   STRING COMMENT '锁定状态修改时间',
    itcast_clazz_id             INT COMMENT '所属ems班级id',
    itcast_clazz_time           STRING COMMENT '报班时间',
    payment_url                 STRING COMMENT '付款链接',
    payment_url_time            STRING COMMENT '支付链接生成时间',
    ems_student_id              INT COMMENT 'ems的学生id',
    delete_reason               STRING COMMENT '删除原因',
    deleter                     INT COMMENT '删除人',
    deleter_name                STRING COMMENT '删除人姓名',
    delete_time                 STRING COMMENT '删除时间',
    course_id                   INT COMMENT '课程ID',
    course_name                 STRING COMMENT '课程名称',
    delete_comment              STRING COMMENT '删除原因说明',
    close_state                 STRING COMMENT '关闭装填',
    close_time                  STRING COMMENT '关闭状态变动时间',
    appeal_id                   INT COMMENT '申诉id',
    tenant                      INT COMMENT '租户',
    total_fee                   DECIMAL COMMENT '报名费总金额',
    belonged                    INT COMMENT '小周期归属人',
    belonged_time               STRING COMMENT '归属时间',
    belonger_time               STRING COMMENT '归属时间',
    transfer                    INT COMMENT '转移人',
    transfer_time               STRING COMMENT '转移时间',
    follow_type                 INT COMMENT '分配类型，0-自动分配，1-手动分配，2-自动转移，3-手动单个转移，4-手动批量转移，5-公海领取',
    transfer_bxg_oa_account     STRING COMMENT '转移到博学谷归属人OA账号',
    transfer_bxg_belonger_name  STRING COMMENT '转移到博学谷归属人OA姓名',
    end_time                    STRING COMMENT '有效截止时间'
) COMMENT '客户关系表' 
PARTITIONED BY (start_time STRING) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' 
STORED AS ORC 
TBLPROPERTIES ('orc.compress' = 'ZLIB');
"

# 四、合并增量数据与历史数据并写到tmp表中
${HIVE_HOME} -e "
--分区
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.exec.max.dynamic.partitions.pernode=10000;
SET hive.exec.max.dynamic.partitions=100000;
SET hive.exec.max.created.files=150000;
--hive压缩
SET hive.exec.compress.intermediate=true;
SET hive.exec.compress.output=true;
--写入时压缩生效
SET hive.exec.orc.compression.strategy=COMPRESSION;
--插入到临时表中，保留备份
INSERT OVERWRITE TABLE itcast_ods.customer_relationship_tmp PARTITION (start_time)
SELECT *
FROM (
         SELECT *
         FROM itcast_ods.customer_relationship_update

         UNION ALL

         SELECT rs.id,
                rs.create_date_time,
                rs.update_date_time,
                rs.deleted,
                rs.customer_id,
                rs.first_id,
                rs.belonger,
                rs.belonger_name,
                rs.initial_belonger,
                rs.distribution_handler,
                rs.business_scrm_department_id,
                rs.last_visit_time,
                rs.next_visit_time,
                rs.origin_type,
                rs.itcast_school_id,
                rs.itcast_subject_id,
                rs.intention_study_type,
                rs.anticipat_signup_date,
                rs.level,
                rs.creator,
                rs.current_creator,
                rs.creator_name,
                rs.origin_channel,
                rs.comment,
                rs.first_customer_clue_id,
                rs.last_customer_clue_id,
                rs.process_state,
                rs.process_time,
                rs.payment_state,
                rs.payment_time,
                rs.signup_state,
                rs.signup_time,
                rs.notice_state,
                rs.notice_time,
                rs.lock_state,
                rs.lock_time,
                rs.itcast_clazz_id,
                rs.itcast_clazz_time,
                rs.payment_url,
                rs.payment_url_time,
                rs.ems_student_id,
                rs.delete_reason,
                rs.deleter,
                rs.deleter_name,
                rs.delete_time,
                rs.course_id,
                rs.course_name,
                rs.delete_comment,
                rs.close_state,
                rs.close_time,
                rs.appeal_id,
                rs.tenant,
                rs.total_fee,
                rs.belonged,
                rs.belonged_time,
                rs.belonger_time,
                rs.transfer,
                rs.transfer_time,
                rs.follow_type,
                rs.transfer_bxg_oa_account,
                rs.transfer_bxg_belonger_name,
                IF(up.id IS NOT NULL AND rs.end_time = '9999-12-31', up.update_date_time, rs.end_time) end_time,
                rs.start_time
         FROM itcast_ods.customer_relationship rs
                  LEFT JOIN
              itcast_ods.customer_relationship_update up
              ON rs.id = up.id
     ) his
ORDER BY his.id, start_time;
"

# 五、临时表覆盖到拉链表
${HIVE_HOME} -e "
--分区
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.exec.max.dynamic.partitions.pernode=10000;
SET hive.exec.max.dynamic.partitions=100000;
SET hive.exec.max.created.files=150000;
--hive压缩
SET hive.exec.compress.intermediate=true;
SET hive.exec.compress.output=true;
--写入时压缩生效
SET hive.exec.orc.compression.strategy=COMPRESSION;
--分桶
SET hive.enforce.bucketing=true;
SET hive.enforce.sorting=true;
SET hive.optimize.bucketmapjoin = true;
SET hive.auto.convert.sortmerge.join=true;
SET hive.auto.convert.sortmerge.join.noconditionaltask=true;
--覆盖写回数据
INSERT OVERWRITE TABLE itcast_ods.customer_relationship PARTITION (start_time)
SELECT *
FROM itcast_ods.customer_relationship_tmp;
"
